package com.xhs.bolt;

import com.alibaba.fastjson.JSONArray;
import com.xhs.utils.HttpClientHelper;
import com.xhs.zk.ZKSession;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.trident.util.LRUMap;
import org.apache.storm.tuple.Tuple;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;

/**
 * Created by jm on 2017/11/20.
 */
public class ProductCountBolt extends BaseRichBolt {

    private static final String LOCK = "TASK-ID";
    private static final String NODEIDSPATH = "/TASK-ID-LIST";
    private static final int TOPN = 3;

    //初始化具有1000长度的LRU算法的map
    private LRUMap<Long,Long> productCount = new LRUMap<>(1000);

    private ZKSession zkSession;

    private int taskId;

    HttpClientHelper httpClientHelper;

    private ArrayBlockingQueue<Long> sortedArray = new ArrayBlockingQueue(1000);

    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.zkSession = ZKSession.getInstance();
        this.taskId = topologyContext.getThisTaskId();
        this.httpClientHelper = HttpClientHelper.getInstance();
        initTaskId();
        new Thread(new SortThread()).start();
        new Thread(new ProductCountThread()).start();
        new Thread(new ProductHotThread()).start();
    }

    private void initTaskId(){

        System.out.println(taskId+" =================initTaskId=================== ");
        //拿到锁
        ZKSession.acquireDistributedLock(LOCK);
        try {
            //创建task list node
            zkSession.createNode(NODEIDSPATH);
        }catch (Exception e){
            System.out.println("已存在 /TASK-ID-LIST 节点");
        }
        //获取一下当前task list node 的值
        String taskIds = zkSession.getNodeData(NODEIDSPATH);

        System.out.println(taskId+"【ProductCountBolt获取到taskid list】taskidList=" + taskIds);
        if (taskIds==null ||taskIds.isEmpty()){
            taskIds += taskId;
        }else {
            taskIds += ","+taskId;
        }
        //拼装后放回task list node 里
        zkSession.setNodeData(NODEIDSPATH,taskIds);

        System.out.println(taskId+"【ProductCountBolt设置taskid list】taskidList=" + taskIds);
        //释放锁
        ZKSession.releaseLock(LOCK);

        System.out.println(taskId+" =================initTaskId=================== ");
    }

    @Override
    public void execute(Tuple tuple) {
        Long productId = tuple.getLongByField("productId");
        System.out.println(taskId+"【ProductCountBolt接收到一个商品id】 productId=" + productId);
        Long count = productCount.get(productId);
        //如果这个产品没有,初始化计数
        if (count == null)
            count = 0L;

        productCount.put(productId,++count);
        System.out.println(taskId+"【ProductCountBolt完成商品访问次数统计】productId=" + productId + ", count=" + count);
    }

    private class ProductHotThread implements Runnable{

        @Override
        public void run() {
            //热门活动
            List<Long> hotProductIds = new ArrayList<>();
            //需要移除的热点
            List<Long> hotRemoveProductIds = new ArrayList<>();
            //本次热点
            List<Long> lastTimeHotProductIds = new ArrayList<>();
            //本次和上次都有的热点
            List<Long> stillHotProductIds = new ArrayList<>();

            while (true) {

                System.out.println();
                System.out.println();
                System.out.println(taskId+" =================ProductHotThread=================== ");
                try {

                    if (productCount.isEmpty() || sortedArray.isEmpty() ){
                        System.out.println(taskId+"【ProductCountBolt 没有热点数据】继续等待.... ");
                        try {
                            Thread.sleep(10 * 1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        continue;
                    }

                    //本次临时热点数组
                    lastTimeHotProductIds.clear();
                    stillHotProductIds.clear();
                    hotRemoveProductIds.clear();
                    //这里有个小技巧,就是指需要排序出前5%的数据就行了,剩下的就是后95%的数据了.
//                    getTopNList(productCount.size(),topnProducts,topnProductids);
                    Long[] sortIds = sortedArray.toArray(new Long[]{});
                    if (sortIds.length < 1){
                        System.out.println(taskId+"【ProductCountBolt 没有排序数据】继续等待.... ");
                        try {
                            Thread.sleep(5 * 1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        continue;
                    }

                    System.out.println(taskId+"【ProductCountBolt 热点数据排序完成】原始数据长度: "+productCount.size()+";  排序后: "+JSONArray.toJSONString(sortIds));

                    if (sortIds.length == 1){
                        //如果就一个数据,就判断单点压力
                        //这里需要提前预估下服务器的承受压力,比如,单节点能承受2w的访问量,那这里就设置1.5w
                        System.out.println(taskId+"【ProductCountBolt】 唯一的一个元素 "+sortIds[0]+" 的访问量: " + productCount.get(sortIds[0]));
                        if (productCount.get(sortIds[0]) > 15000){
                            System.out.println(taskId+"【ProductCountBolt 单个热点】topnProductids: "+JSONArray.toJSONString(sortIds));
                            /**
                             * 这里需要做两件事
                             * 1 给分发nginx发送变更策略的请求
                             * 2 给应用nginx发送活动数据,让应用服务器缓存到本地.
                             * 这里就不实现了.
                             */
                        }else {
                            try {
                                Thread.sleep(10 * 1000);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                            continue;
                        }
                    }
                    /**
                     * 测算热点数据的方法有多种.通常可以按
                     * 1 前5%的访问量如果是后95%均值的5倍以上,就可以认为是热点数据.
                     * 2 前20%的数据就认为是热点数据, 这里也可以按2/8原则,20%的产品访问量展整个网站压力的80%,所以可以通过排序后把前20%的产品做为热点数据处理,这里就不演示了.
                     */

                    //这里按第一种.计算数组 后95% 的数据有多少,
                    int index95 = (int) Math.floor(sortIds.length * 0.7);
                    System.out.println(taskId+"【ProductCountBolt 热点数据 index95 : "+index95);

//                    Long[] ids = topnProductids.toArray(new Long[]{});

                    long avgCount = 0;

                    //后95%的访问次数
                    for (int i = sortIds.length ; i > sortIds.length - index95 ; i -- ){
                        //累加 后95% 下标对应的访问次数
                        avgCount += productCount.get(sortIds[i-1]);
                    }
                    System.out.println(taskId+"【ProductCountBolt 热点数据 后95%访问量: "+avgCount);
                    //后95%访问量的均值
                    avgCount  =  avgCount / index95;
                    System.out.println(taskId+"【ProductCountBolt 热点数据 后95%访问量的均值: "+avgCount);

                    //前5%的访问量 如果是后95%访问量均值的5倍,则认为是热点数据.
                    /**
                     * TODO 优化点,拿前5%的数据的最后一个下标逐个往前,和均值进行比较,如果某个下标的访问量是后95%均值的5倍,那么这个下标之前所有的数据都可以认为是热点数据,没必要全部比较一次.
                     * eg:avgCount = 2,那么5倍就是10
                     *
                     * 如果前5%的sortIds 为   9   8   2   7   3
                     * 对应的访问量分别是     50  30  11   9   7
                     *
                     * 那么从id为3 开始倒着比较,当比较到id = 2的时候,发现2的访问量是11,大于10,那么就认为id为2 8 9的product都为热点活动.
                     *
                     */
                    for (int i = 0; i < sortIds.length - index95 ; i ++ ){
                        long tmpCount = productCount.get(sortIds[i]);
                        if (tmpCount > avgCount * 5){
                            lastTimeHotProductIds.add(sortIds[i]);
                        }
                    }
                    System.out.println(taskId+"【ProductCountBolt 算出热点数据: "+JSONArray.toJSONString(lastTimeHotProductIds));

                    if (!lastTimeHotProductIds.isEmpty()) {
                        //如果热点数组为空,就全加入到热点数组
                        if (hotProductIds.isEmpty()) {
                            hotProductIds.addAll(lastTimeHotProductIds);
                        }else {
                            for (Long id : lastTimeHotProductIds){
                                //上次的热点 在这次热点中没有,就说明已经不是热点了.就移除掉
                                if(!hotProductIds.contains(id)){
                                    hotRemoveProductIds.add(id);
                                }else {
                                    //如果找的到,说明上次的热点依然有效,那么就不需要再次查询热点的相关信息了
                                    stillHotProductIds.add(id);
                                }
                            }
                            //找到需要移除的热点后,把这次的热点加入热点数组中.
                            hotProductIds.clear();
                            hotProductIds.addAll(lastTimeHotProductIds);
                        }


                        String routeNginxUrl = "http://101.200.183.100:8888/hot?productId=";

                        String appNginxUrl1 = "http://101.200.169.10:8888/hot?productId=";
                        String appNginxUrl2 = "http://101.201.101.50:8888/hot?productId=";
                        String productUrl = "http://101.200.183.100:9999/getProductInfo/";

                        System.out.println(taskId+"【ProductCountBolt 依然是热点: "+JSONArray.toJSONString(stillHotProductIds));
                        //增加热点
                        for(Long id : hotProductIds){

                            //推送到各应用nginx 产品完整信息,让各个应用nginx先缓存到本地
                            if (!stillHotProductIds.contains(id)) {
                                //如果当前的热点 已经在之前的热点里面,就不需要再次查产品信息了.(但是这里需要注意,如果nginx中的缓存到期了,该做相应的处理)
                                String respProductInfo = httpClientHelper.reqGetResponse(productUrl+id);
                                System.out.println(taskId+"【ProductCountBolt 获取到商品信息:"+respProductInfo);
                                if (respProductInfo != null) {

                                    httpClientHelper.reqGet(appNginxUrl1 + id + "&productInfo=");
                                    httpClientHelper.reqGet(appNginxUrl2 + id + "&productInfo=");
                                }
                                System.out.println(taskId+"[成功] 请求发送 --->  应用nginx " + id);

                                //推送到分发nginx产品的id,修改分发策略.为负载均衡路由策略
                                httpClientHelper.reqGet(routeNginxUrl+id);
                                System.out.println(taskId+"[成功] 请求发送 --->  分发nginx "+id);
                            }else {
                                System.out.println(taskId+"[成功] 依然是热点 --->  分发nginx "+id);
                                System.out.println(taskId+"[成功] 依然是热点 --->  应用nginx "+id);
                            }

                        }

                    }else {
                        System.out.println(taskId+"【ProductCountBolt 清空所有热点数据");
                        //如果这次没有热点,就将所有热点清楚
                        hotRemoveProductIds.addAll(hotProductIds);
                        hotProductIds.clear();
                    }


                    String routeRemoveNginxUrl = "http://101.200.183.100:8888/cancel?productId=";
                    System.out.println(taskId+"【ProductCountBolt 即将移除的热点: "+JSONArray.toJSONString(hotRemoveProductIds));
                    //移除热点
                    for(Long id : hotRemoveProductIds){
                        //推送到分发nginx产品的id,修改分发策略.为hash定向路由策略
                        httpClientHelper.reqGet(routeRemoveNginxUrl+id);
                        System.out.println(taskId+"[成功] 移除热点 --->  分发nginx "+id);
                    }


                    System.out.println(taskId+" =================ProductHotThread=================== ");
                    Thread.sleep(10 * 1000);

                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        }
    }



    /**
     * 在LRUMap中排序,获取
     */
    private class ProductCountThread implements Runnable{

        @Override
        public void run() {
            //Top N 热点活动列表
//            List<Map.Entry<Long,Long>> topnProducts = new ArrayList<>();
            //Top N 热点活动id列表
            Long[] topnProductids = new Long[TOPN];
            while (true) {
                System.out.println();
                System.out.println();
                System.out.println(taskId+" =================ProductCountThread=================== ");
                try {
                    if (!sortedArray.isEmpty()) {
                        Long[] sortIds = sortedArray.toArray(new Long[]{});
                        for (int i = 0; i < TOPN; i++) {
                            if (i >= sortIds.length)
                                break;
                            topnProductids[i] = sortIds[i];
                        }
                        zkSession.createNode("/task-hot-product-ids-" + taskId);
                        zkSession.setNodeData("/task-hot-product-ids-" + taskId, JSONArray.toJSONString(topnProductids));

                        System.out.print(taskId + "【ProductCountThread计算出一份top3热门商品列表】zk path=" + ("/task-hot-product-list-" + taskId) + ", product-ids=" + JSONArray.toJSONString(topnProductids) + "\n\n");
                    }
                    System.out.println(taskId+" =================ProductCountThread=================== ");
                    Thread.sleep(20 * 1000);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 定时(10s)排序线程
     */
    private class SortThread implements Runnable{
        @Override
        public void run() {
            while (true)
            try {
                System.out.println();
                System.out.println();
                System.out.println(taskId+" =================SortThread=================== ");
                //每次清空排序数据
                sortedArray.clear();

                Long[] ids = getTopNList();
                if (ids != null && ids.length > 0){
                    for (long id : ids){
                        //want not block
                        sortedArray.offer(id);
                    }
                }

                System.out.println(taskId+" =================SortThread=================== ");
                Thread.sleep(10 * 1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }


    /**
     * 排序
     * @return 排序后的 productCount key 组成的 key 数组
     */
//    private List<Long> getTopNList(int topn,List<Map.Entry<Long,Long>> topnProducts,List<Long> topnProductids) {
    private Long[] getTopNList() {

        if (!productCount.isEmpty()) {
            Long[] ids  = productCount.keySet().toArray(new Long[]{});
//            System.out.println(taskId+"【SortThread getTopNList 排序前 】" + Arrays.asList(ids));
            //插入排序 start
            // 假定第 0 个下标是有序的.此后不断的想前比较和插入
            //eg:
            // [3][6][7][9][10]
            for (int i=1 ; i< ids.length ; i++){
                //这里需要用倒序
                //如果后一个元素数据 > 前一个
                if (productCount.get(ids[i]) >  productCount.get(ids[i-1])){
                    //将当前(大于前一个)的 下标j,值id,对应的数据tmpCount 拿出来
                    //eg:
                    // [3][6][7][9][10]
                    // j = 1
                    // id = 6
                    // tmpCount = xxx 具体的数值.普通插入排序不需要考虑这个值,直接拿 下标对应的数据进行比较就OK了.
                    int j = i;
                    long tmpCount = productCount.get(ids[j]);
                    long id = ids[j];
                    //拿当前值挨个和前面比较,如果当前的值 > 前一个就把前一个向后挪动,
                    //因为已经拿出当前的值,所以位置会空出来
                    //eg:
                    // [3][ ][7][9][10]
                    while (j > 0 && tmpCount > productCount.get(ids[j-1])){
                        ids[j] = ids[j-1];//后挪
                        //eg:
                        // [ ][3][7][9][10]
                        j--;
                    }
                    ids[j] = id;
                    //eg:
                    // [6][3][7][9][10]
                }
            }

            System.out.println(taskId+"【SortThread getTopNList 排序完成】" + JSONArray.toJSONString(ids));

            return ids;

        }

        return null;
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }
}
